package yo.latyoo.hokage.skmr.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import yo.latyoo.hokage.skmr.biz.model.JobModel;
import yo.latyoo.hokage.skmr.biz.repository.JobInfoRepo;
import yo.latyoo.hokage.skmr.configuration.SkmrConfiguration;
import yo.latyoo.hokage.skmr.trigger.JobTrigger;
import yo.latyoo.hokage.tools.consts.LogConst;
import yo.latyoo.hokage.tools.exception.HokageException;
import yo.latyoo.hokage.tools.expression.cron.CronExpression;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 任务调度触发入口类
 */
public class JobSchedulerThread {
    private JobSchedulerThread(){}
    private static final Logger logger = LoggerFactory.getLogger(JobSchedulerThread.class);
    private static final JobSchedulerThread instance = new JobSchedulerThread();
    public static JobSchedulerThread INSTANCE(){
        return instance;
    }
    //定时任务轮询守护线程  不间断从数据库读取任务
    private Thread scheduleThread;
    //时间轮守护线程
    private Thread timeWheelThread;
    private volatile boolean schedulerStop = false;
    private volatile boolean timeWheelStop = false;
    public static final long PRE_READ_SECONDS = 5;    // 预读5s

    //时间轮
    private volatile static Map<Integer, List<JobModel>> TIME_WHEEL_MAP = new ConcurrentHashMap<>();

    public void start(){
        scheduleThread = new Thread(() -> {
            SkmrConfiguration skmrConfiguration = SkmrConfiguration.getConfig();
            JobInfoRepo jobInfoRepo = skmrConfiguration.getJobInfoRepo();

            DataSource dataSource = skmrConfiguration.getDataSource();
            while (!schedulerStop) {
                long start = System.currentTimeMillis();
                Connection conn = null;
                Boolean connAutoCommit = null;
                PreparedStatement preparedStatement = null;
                LocalDateTime nowTime = LocalDateTime.now();
                boolean preReadSuc = true;
                try {
                    conn = dataSource.getConnection();
                    connAutoCommit = conn.getAutoCommit();
                    conn.setAutoCommit(false);
                    //获取数据库排他锁，多个调度服务在同一时间只能有一个服务获取资源（前提是多个调度服务链接一个数据库）
                    preparedStatement = conn.prepareStatement("select * from  job_lock where lock_name = 'schedule_lock' for update");
                    preparedStatement.execute();
                    //查询出小于当前时间+5s的数据
                   List<JobModel> infoList = jobInfoRepo.findByNextRunTime(nowTime.plusSeconds(PRE_READ_SECONDS));
                   if(infoList.size() > 0)
                       try(var executorService = Executors.newVirtualThreadPerTaskExecutor()){
                            infoList.forEach(it->{
                                executorService.submit(()->{
                                    //获取到的任务 分三种执行策略
                                    if(it.nextRunTime.plusSeconds(PRE_READ_SECONDS).isBefore(nowTime)){
                                        //1、已经过期5s以上，直接更新下一次执行时间
                                        generatorNextTime(it,LocalDateTime.now());
                                    }
                                    //执行时间小于当前时间
                                    else if(it.nextRunTime.isBefore(nowTime)){
                                        //2、需要执行的正常数据，然后更新下一次执行时间
                                        JobTrigger.trigger(it);
                                        generatorNextTime(it,LocalDateTime.now());
                                        if(it.nextRunTime.isBefore(nowTime.plusSeconds(PRE_READ_SECONDS))){
                                            pushTimingWheel(it.nextRunTime.getSecond(),it);
                                            generatorNextTime(it,it.nextRunTime);
                                        }
                                    }else{
                                        //3、还没到执行时间，则放入是时间轮中调度,并更新下一次执行时间
                                        pushTimingWheel(it.nextRunTime.getSecond(),it);
                                        generatorNextTime(it,it.nextRunTime);
                                    }
                                    jobInfoRepo.updateNextRunTime(it.id,it.nextRunTime,it.lastRunTime, LocalDateTime.now());
                                });

                            });
                        }


                    if(infoList.size() < 1){
                        preReadSuc = false;
                    }

                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    if (conn != null) {
                        try {
                            conn.commit();
                            conn.setAutoCommit(Boolean.TRUE.equals(connAutoCommit));
                            conn.close();
                        } catch (SQLException e) {
                            if (!schedulerStop) {
                                logger.error(LogConst.PREFIX + e.getMessage(), e);
                            }
                        }
                    }
                }
                long cost = System.currentTimeMillis()-start;
                // Wait seconds, align second
                if (cost < 1000) {  // scan-overtime, not wait
                    try {
                        // pre-read period: success > scan each second; fail > skip this period;
                        TimeUnit.MILLISECONDS.sleep((preReadSuc?1:PRE_READ_SECONDS)*1000 - System.currentTimeMillis()%1000);
                    } catch (InterruptedException e) {
                        if (!schedulerStop) {
                            logger.error(LogConst.PREFIX + e.getMessage(), e);
                        }
                    }
                }

            }
        });

        scheduleThread.setDaemon(true);
        scheduleThread.setName("latyoo-hokage-skmr, JobScheduleThread#scheduleThread");
        scheduleThread.start();

        timeWheelThread = new Thread(()->{
            while (!timeWheelStop) {
                try {
                    TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
                } catch (InterruptedException e) {
                    if (!timeWheelStop) {
                        logger.error(LogConst.PREFIX + e.getMessage(), e);
                    }
                }

                try {
                    // second data
                    List<JobModel> ringItemData = new ArrayList<>();
                    int nowSecond = LocalDateTime.now().getSecond();   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
                    for (int i = 0; i < 2; i++) {
                        List<JobModel> tmpData = TIME_WHEEL_MAP.remove((nowSecond + 60 - i) % 60);
                        if (tmpData != null) {
                            ringItemData.addAll(tmpData);
                        }
                    }

                    // ring trigger
                    logger.debug(LogConst.PREFIX + "pull timeWheel data to execute : " + nowSecond + " = " + List.of(ringItemData));
                    if (ringItemData.size() > 0) {
                        // do trigger
                        for (JobModel job : ringItemData) {
                            JobTrigger.trigger(job);
                        }
                        // clear
                        ringItemData.clear();
                    }
                } catch (Exception e) {
                    if (!timeWheelStop) {
                        logger.error(LogConst.PREFIX + "JobScheduleHelper#ringThread error.", e);
                    }
                }
            }
        });

        timeWheelThread.setDaemon(true);
        timeWheelThread.setName("latyoo-hokage-skmr, JobScheduleThread#timeWheelThread");
        timeWheelThread.start();
    }


    public void stop(){
        instance.toStop();
    }
    private  void toStop(){
        // 1、stop schedule
        schedulerStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);  // wait
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
        if (scheduleThread.getState() != Thread.State.TERMINATED){
            // interrupt and wait
            scheduleThread.interrupt();
            try {
                scheduleThread.join();
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        boolean hasRingData = false;
        if (!TIME_WHEEL_MAP.isEmpty()) {
            for (int second : TIME_WHEEL_MAP.keySet()) {
                List<JobModel> tmpData = TIME_WHEEL_MAP.get(second);
                if (tmpData!=null && tmpData.size()>0) {
                    hasRingData = true;
                    break;
                }
            }
        }
        if (hasRingData) {
            try {
                TimeUnit.SECONDS.sleep(8);
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        // stop ring (wait job-in-memory stop)
        timeWheelStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
        if (timeWheelThread.getState() != Thread.State.TERMINATED){
            // interrupt and wait
            timeWheelThread.interrupt();
            try {
                timeWheelThread.join();
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }
        logger.info("{} JobScheduleThread stop.",LogConst.PREFIX);
    }

    private void generatorNextTime(JobModel info, LocalDateTime fromTime)  {
        ZoneId zoneId = ZoneId.of("Asia/Shanghai");
        ZonedDateTime zdt = fromTime.atZone(zoneId);
        Date date = Date.from(zdt.toInstant());
        try {
            LocalDateTime nextValidTimeAfter = switch (info.scheduleType) {
                case CRON -> new CronExpression(info.expression)
                        .getNextValidTimeAfter(date)
                        .toInstant().atZone(zoneId).toLocalDateTime();
                case FIX_RATE -> fromTime.plusSeconds(Long.parseLong(info.expression));
            };
            info.lastRunTime = info.nextRunTime;
            info.nextRunTime = nextValidTimeAfter;
        }catch (Exception e){
            logger.error(LogConst.PREFIX + "jobId:{}, expression error",info.id,e);
            throw new HokageException(e.getMessage());
        }
    }

    private void pushTimingWheel(int seconds,JobModel jobId){
        TIME_WHEEL_MAP.computeIfAbsent(seconds,it -> new ArrayList<>()).add(jobId);
    }

}
